/*
* Copyright 2008 Lukas Vlcek
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

package org.permutation;

import java.util.ArrayList;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.mapred.jobcontrol.Job;
import org.apache.hadoop.mapred.jobcontrol.JobControl;

/**
 * The main controller for permutation generation process.
 * It needs a feed which is then developed into a new generation.
 * This process is repeted in cycles. The number of cycles is determined
 * by internal variable <code>cycles</code>. The feed generation is identified
 * by a number stored in other internal variable <code>start</code>.
 * Input and output folders (feed permutation and new permuattion generation)
 * has a name <code>jobG</code> where <code>G</code> is a number of generation.
 * So typically there needs to be at least one feed folder <code>job1</code>
 * which contains a file with the first permutation generation (in this case
 * it is just a file <code>part-00000</code> with only one row containing
 * the string "1").
 * <p>
 * TODO: There are way to much hardcoded things in the code, make it more
 * configurable.
 * 
 * @author Lukas Vlcek
 */
public class Permutation {

    public static void main(String[] args) throws Exception {

        int start = 1;
        int cycles = 7;

        JobControl control = new JobControl("Permutation");

        ArrayList<Job> dependingJobs = null;
        Job lastJob = null;

        for (int i = 1; i <= cycles; i++) {

            dependingJobs = new ArrayList<Job>();
            if (lastJob != null) {
                dependingJobs.add(lastJob);
            }

            Configuration defaults = new Configuration();
            JobConf conf = new JobConf(defaults, Permutation.class);

            Path input = new Path("job" + (start+i-1) + "/part-00000");
            Path output = new Path("job" + (start+i));

            FileInputFormat.setInputPaths(conf, input);
            FileOutputFormat.setOutputPath(conf, output);

            conf.setOutputKeyClass(Text.class);
            conf.setOutputValueClass(Text.class);

            conf.setMapperClass(PermutationMap.class);
            //conf.setCombinerClass(PermutationReduce.class);
            conf.setReducerClass(PermutationReduce.class);

            conf.setInputFormat(TextInputFormat.class);
            conf.setOutputFormat(TextOutputFormat.class);

            conf.setNumMapTasks((start+i+1)*4);
            conf.setNumReduceTasks(start+i+1);

            Job actualJob = new Job(conf, dependingJobs);
            actualJob.setJobName("PermutationGen" + (start+i));

            control.addJob(actualJob);

            lastJob = actualJob;
        }


        Thread controller = new Thread(control);
        controller.start();

        while (!control.allFinished()) {
            try {
                /*
                System.out.println("Jobs in waiting state: " + control.getWaitingJobs().size());
                System.out.println("Jobs in ready state: " + control.getReadyJobs().size());
                System.out.println("Jobs in running state: " + control.getRunningJobs().size());
                System.out.println("Jobs in success state: " + control.getSuccessfulJobs().size());
                System.out.println("Jobs in failed state: " + control.getFailedJobs().size());
                System.out.println("\n");
                 
                Thread.sleep(2000);
                 * */
            } catch (Exception e) {
            }
        }
        control.stop();
        System.out.println("done.");
    }
}
